#pragma once
#include "abstract.hpp"

#include "muduobase/include/muduo/net/TcpServer.h"
#include "muduobase/include/muduo/net/EventLoop.h"
#include "muduobase/include/muduo/net/TcpConnection.h"
#include "muduobase/include/muduo/net/Buffer.h"
#include "muduobase/include/muduo/base/CountDownLatch.h"
#include "muduobase/include/muduo/net/EventLoopThread.h"
#include "muduobase/include/muduo/net/TcpClient.h"
#include "field.hpp"
#include "tool.hpp"
#include "message.hpp"
#include <arpa/inet.h>
#include <unordered_map>
#include <mutex>

namespace RPC
{
    class MuduoBuffer : public BaseBuffer
    {
    public:
        using Ptr = std::shared_ptr<MuduoBuffer>;
        MuduoBuffer() : _buf(nullptr) {}
        MuduoBuffer(muduo::net::Buffer *buf) : _buf(buf) {}
        virtual size_t readAbleSize() override
        {
            return _buf->readableBytes();
        }

        virtual int32_t peekInt32_t() override
        {
            return _buf->peekInt32();
        }

        virtual void retrieveInt32_t() override
        {
            return _buf->retrieveInt32();
        }

        virtual int32_t readInt32_t() override
        {
            return _buf->readInt32();
        }

        virtual std::string retrieveAsString(size_t len) override
        {
            return _buf->retrieveAsString(len);
        }

    private:
        muduo::net::Buffer *_buf;
    };

    class BufferFactory
    {
    public:
        template <typename... Args>
        static BaseBuffer::Ptr create(Args &&...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    class LvProtocol : public BaseProtocol
    {
    public:
        // |---4 totallen---|---4 type---|---4 idlen---|---id---|---body---|
        using Ptr = std::shared_ptr<LvProtocol>;
        virtual bool canProcessd(const BaseBuffer::Ptr &buffer) override
        {
            int rd_able_size = buffer->readAbleSize();
            // std::cout << "rd_able_size: " << rd_able_size << 
            // std::endl;
            // std::cout << "lenFieldLength: " << lenFieldsLength << std::endl;
            if (rd_able_size < lenFieldsLength)
            {
                // ERROR_LOG("here1");
                return false;
            }
            int len = buffer->peekInt32_t() + lenFieldsLength;
            // std::cout << "len: " << len << std::endl;
            if (rd_able_size < len)
            {
                // ERROR_LOG("here2");
                return false;
            }
            return true;
        }

        // 这一步其实就是对LV格式反序列化
        virtual bool onMessage(const BaseBuffer::Ptr &buffer, BaseMessage::Ptr &message) override
        {
            // 读出每个字段的长度,拿到body
            // 我们在实现中也有反序列化，那个是针对body
            int totallen = buffer->readInt32_t();
            int type = buffer->readInt32_t();
            int idlen = buffer->readInt32_t();
            std::string id = buffer->retrieveAsString(idlen);

            INFO_LOG("--------receive msg-----------");
            std::cout << " totallen: " << totallen << std::endl;
            std::cout << " type: " << mType((MType)type) << std::endl;
            std::cout << "idlen: " << idlen << std::endl;
            int bodylen = totallen - typeFieldsLength - idFieldsLength - idlen;
            std::string body = buffer->retrieveAsString(bodylen);

            message = MessageFactory::create((RPC::MType)type);
            if (message == nullptr)
            {
                ERROR_LOG("MessageFactory creates failed!");
            }
            // 这里的反序列化将字符串序列号放在json key_val格式里面
            bool ret = message->desrialize(body);
            if (ret == false)
            {
                ERROR_LOG("Message deserialization failed!");
                return false;
            }
            // std::cout << message->_body.toStyledString();
            message->setId(id);
            message->setType((RPC::MType)type);
            return true;
        }

        // 这里是将LV格式序列化，底层message自己的序列化与这里无关
        virtual std::string serialize(const BaseMessage::Ptr &message)
        {
            int type = (int)message->type();
            std::string id = message->id();
            std::string body = message->serialize();
            // 转成网络字节序。
            int type_n = htonl(type);
            int idlen_n = htonl(id.size());
            int totallen_h = typeFieldsLength + idFieldsLength + id.size() + body.size();
            int totallen_n = htonl(totallen_h);
            // 序列化形成字符串
            std::string ret;
            ret.reserve(totallen_h + lenFieldsLength);
            ret.append((char *)&totallen_n, lenFieldsLength);
            ret.append((char *)&type_n, typeFieldsLength);
            ret.append((char *)&idlen_n, idFieldsLength);
            ret.append(id);
            ret.append(body);
            // std::cout << "type: " << type << std::endl;
            // std::cout << "id.size: " << id.size() << std::endl;
            // std::cout << "body.size(): " << body.size() << std::endl;
            // std::cout << "totallen_h" << totallen_h << std::endl;
            // ERROR_LOG("%s", id.c_str());
            // ERROR_LOG("%s", body.c_str());
            return ret;
        }

    private:
        size_t lenFieldsLength = 4;
        size_t typeFieldsLength = 4;
        size_t idFieldsLength = 4;
    };

    class ProtocolFactory
    {
    public:
        template <typename... Args>
        static BaseProtocol::Ptr create(Args &&...args)
        {
            return std::make_shared<LvProtocol>(std::forward<Args>(args)...);
        }
    };

    class MuduoConnection : public BaseConnection
    {
    public:
        using Ptr = std::shared_ptr<MuduoConnection>;

        MuduoConnection(const muduo::net::TcpConnectionPtr &conn,
                        const BaseProtocol::Ptr &protocol)
            : _protocol(protocol), _conn(conn)
        {
        }

        virtual void send(const BaseMessage::Ptr &msg) override
        {
            std::string body = _protocol->serialize(msg);
            std::cout << body.c_str() << std::endl;
            // DEBUG_LOG("%s", body.c_str());
            _conn->send(body);
        }

        virtual bool connected() override
        {
            return _conn->connected();
        }

        virtual void shutdown() override
        {
            return _conn->shutdown();
        }

    private:
        BaseProtocol::Ptr _protocol;
        muduo::net::TcpConnectionPtr _conn;
    };

    class ConnectionFactory
    {
    public:
        template <typename... Args>
        static BaseConnection::Ptr create(Args &&...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    class MuduoServer : public BaseServer
    {
    public:
        using Ptr = std::shared_ptr<MuduoServer>;

        MuduoServer(uint16_t port)
            : _protocol(RPC::ProtocolFactory::create()),
              _server(&_evtloop, muduo::net::InetAddress("0.0.0.0", port),
                      "MuduoServer", muduo::net::TcpServer::kReusePort)
        {
        }

        virtual void start() override
        {
            _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&MuduoServer::onMessage, this, std::placeholders::_1,
                                                 std::placeholders::_2, std::placeholders::_3));
            _server.start(); // 监听
            INFO_LOG("开始事件循环监控！");
            _evtloop.loop(); // 事件循环监控
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected()) // 连接建立
            {
                INFO_LOG("连接断开！");
                BaseConnection::Ptr base_conn = ConnectionFactory::create(conn, _protocol);
                {
                    std::unique_lock<std::mutex> _lock(_mutex);
                    _conns.insert({conn, base_conn});
                }
                if (_cb_connecton)
                    _cb_connecton();
            }
            else // 连接断开
            {
                INFO_LOG("连接断开！");
                BaseConnection::Ptr muduo_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        return;
                    }
                    muduo_conn = it->second;
                    _conns.erase(conn);
                }
                if (_cb_close)
                    _cb_close(muduo_conn);
            }
        }

        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buff, muduo::Timestamp)
        {
            INFO_LOG("有消息来到了！");
            BaseBuffer::Ptr base_buff = BufferFactory::create(buff);
            while (1)
            {
                bool ret = _protocol->canProcessd(base_buff);
                if (ret == false)
                {
                    if (base_buff->readAbleSize() > _max_buffer_size)
                    {
                        ERROR_LOG("恶意消息，取消接收！");
                        conn->shutdown();
                        return;
                    }
                    INFO_LOG("消息长度不够！");
                    break;
                }
                BaseMessage::Ptr base_msg;
                ret = _protocol->onMessage(base_buff, base_msg);
                if (ret == false)
                {
                    ERROR_LOG("消息反序列化失败！");
                    conn->shutdown();
                    return;
                }
                BaseConnection::Ptr base_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        conn->shutdown();
                        ERROR_LOG("哈希表找不到对应，baseconn!");
                        return;
                    }
                    base_conn = it->second;
                }
                if (_cb_message)
                    _cb_message(base_conn, base_msg);
            }
        }

    private:
        const size_t _max_buffer_size = 1 << 16;
        BaseProtocol::Ptr _protocol;
        muduo::net::EventLoop _evtloop;
        muduo::net::TcpServer _server;
        std::mutex _mutex;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::Ptr> _conns;
    };

    class ServerFactory
    {
    public:
        template <typename... Args>
        static BaseServer::Ptr create(Args &&...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    class MuduoClient : public BaseClient
    {
    public:
        using MuduoClientPtr = std::shared_ptr<MuduoClient>;
        MuduoClient(const std::string &ip, const uint16_t &port)
            : _protocol(ProtocolFactory::create()), _baseloop(_loopthread.startLoop()),
              _downlatch(1),
              _client(_baseloop, muduo::net::InetAddress(ip, port), "MuduoClient")
        {
        }

        virtual void connect() override
        {
            // DEBUG_LOG("设置回调函数！");
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection,
                                                    this, std::placeholders::_1));
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this,
                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _client.connect();
            _downlatch.wait();
            INFO_LOG("服务器连接成功！");
        }

        bool connected() override
        {
            return _conn->connected();
        }
        void shutdown() override
        {
            _conn->shutdown();
        }
        void send(const BaseMessage::Ptr &msg) override
        {
            std::string msgstr = _protocol->serialize(msg);
            _conn->send(msg);
        }

        BaseConnection::Ptr conn()
        {
            return _conn;
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                DEBUG_LOG("连接成功！");
                _downlatch.countDown();
                _conn = ConnectionFactory::create(conn, _protocol);
            }
            else
            {
                DEBUG_LOG("连接断开！");
                _conn.reset();
            }
        }

        void onMessage(const muduo::net::TcpConnectionPtr &conn,
                       muduo::net::Buffer *buff,
                       muduo::Timestamp)
        {
            DEBUG_LOG("有信息到来了！");
            BaseBuffer::Ptr base_buff = BufferFactory::create(buff);
            while (1)
            {
                if (_protocol->canProcessd(base_buff) == false)
                {
                    // 数据不足
                    DEBUG_LOG("数据不足！");
                    break;
                }
                BaseMessage::Ptr base_msg;
                bool ret = _protocol->onMessage(base_buff, base_msg);
                if (ret == false)
                {
                    DEBUG_LOG("数据类型错误,解析失败！");
                    break;
                }

                if (_cb_message)
                    _cb_message(_conn, base_msg);
            }
        }

    private:
        const size_t _max_buffer_size = 1 << 16;
        BaseProtocol::Ptr _protocol;
        BaseConnection::Ptr _conn;
        muduo::CountDownLatch _downlatch;
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop *_baseloop;
        muduo::net::TcpClient _client;
    };

    class ClientFactory
    {
    public:
        template <typename... Args>
        static BaseClient::Ptr create(Args &&...args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };
}